package com.github.ghsea.scheduler.zkclient;

import java.io.IOException;
import java.util.List;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.leader.LeaderLatch;
import org.apache.curator.framework.recipes.leader.LeaderLatchListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.CloseableExecutorService;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException.NoNodeException;
import org.apache.zookeeper.KeeperException.NodeExistsException;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.util.concurrent.ThreadFactoryBuilder;

public class ZkClient {

	private static ConcurrentHashMap<String, ZkClient> zkServers2Client = new ConcurrentHashMap<>(1);

	private ConcurrentHashMap<String, PathChildrenCache> path2ChildrenCache = new ConcurrentHashMap<>();

	private ConcurrentHashMap<String, NodeCache> path2NodeCache = new ConcurrentHashMap<>();

	private CloseableExecutorService executor;

	private Serializer serializer = new JDKSerializer();

	private static Object lock = new Object();

	private Logger log = LoggerFactory.getLogger(ZkClient.class);

	private CuratorFramework client;

	private ZkClient(String zkServers) {
		RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
		client = CuratorFrameworkFactory.newClient(zkServers, retryPolicy);
		client.start();

		ThreadFactory tf = new ThreadFactoryBuilder().setNameFormat("ZkClient Listener thread factory %d").build();
		executor = new CloseableExecutorService(Executors.newSingleThreadExecutor(tf));
	}

	public void close() {
		path2ChildrenCache.entrySet().forEach(e -> {
			try {
				e.getValue().close();
			} catch (IOException e1) {
				log.error(e1.getMessage(), e1);
			}
		});

		path2NodeCache.entrySet().forEach(e -> {
			try {
				e.getValue().close();
			} catch (IOException e1) {
				log.error(e1.getMessage(), e1);
			}
		});

		if (executor != null) {
			executor.close();
		}

		Set<Entry<String, ZkClient>> allEntry = zkServers2Client.entrySet();
		for (Entry<String, ZkClient> each : allEntry) {
			ZkClient zk = each.getValue();
			if (CuratorFrameworkState.STARTED.equals(zk.client.getState())) {
				zk.client.close();
			}
		}
	}

	public void setSerializer(Serializer serializer) {
		Objects.requireNonNull(serializer, "serializer can'b be null");
		this.serializer = serializer;
	}

	public static ZkClient getInstance(String zkServers) {
		ZkClient zkClient = zkServers2Client.get(zkServers);
		if (zkClient != null) {
			return zkClient;
		}

		synchronized (lock) {
			if (zkClient == null) {
				zkClient = new ZkClient(zkServers);
				zkServers2Client.putIfAbsent(zkServers, zkClient);
			}
		}

		return zkClient;
	}

	public void create(String path) throws Exception {
		create(path, null);
	}

	public void create(String path, Object data) throws Exception {
		byte[] bytDate = serializer.serialize(data);
		client.create().creatingParentsIfNeeded().forPath(path, bytDate);
	}

	public void createEphemeral(String path) throws Exception {
		createEphemeral(path, null, false);
	}

	public void createEphemeral(String path, Object data) throws Exception {
		createEphemeral(path, data, false);
	}

	/**
	 * @param path
	 * @param data
	 * @param rebuildIfReconnected session过期后成功重连时，是否自动重建该节点.在服务发现的场景下，自动重建很有必要。
	 * @throws Exception
	 */
	public void createEphemeral(String path, Object data, boolean rebuildIfReconnected) throws Exception {
		try {
			byte[] bytDate = serializer.serialize(data);
			client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(path, bytDate);
			if (rebuildIfReconnected) {
				client.getConnectionStateListenable().addListener(new EphemeralNodeRebuilder(this, path, data));
			}
		} catch (Exception ex) {
			if (!(ex instanceof NodeExistsException)) {
				throw ex;
			}
		}
	}

	public String createEphemeralSequential(String path, Object data) throws Exception {
		// this will create the given EPHEMERAL-SEQUENTIAL ZNode with the given data
		// using Curator protection.

		/*
		 * Protection Mode:
		 *
		 * It turns out there is an edge case that exists when creating
		 * sequential-ephemeral nodes. The creation can succeed on the server, but the
		 * server can crash before the created node name is returned to the client.
		 * However, the ZK session is still valid so the ephemeral node is not deleted.
		 * Thus, there is no way for the client to determine what node was created for
		 * them.
		 *
		 * Even without sequential-ephemeral, however, the create can succeed on the
		 * sever but the client (for various reasons) will not know it. Putting the
		 * create builder into protection mode works around this. The name of the node
		 * that is created is prefixed with a GUID. If node creation fails the normal
		 * retry mechanism will occur. On the retry, the parent path is first searched
		 * for a node that has the GUID in it. If that node is found, it is assumed to
		 * be the lost node that was successfully created on the first try and is
		 * returned to the caller.
		 */
		byte[] bytDate = serializer.serialize(data);
		return client.create().withProtection().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(path, bytDate);
	}

	public void setData(String path, Object data) throws Exception {
		byte[] bytDate = serializer.serialize(data);
		client.setData().forPath(path, bytDate);
	}

	public Object getData(String path) throws Exception {
		byte[] data = client.getData().forPath(path);
		return serializer.deserialize(data);
	}

	public Object getDataAndWatch(String path, NodeChangedListener listener) throws Exception {
		NodeCache cache = path2NodeCache.get(path);
		if (cache == null) {
			synchronized (path2NodeCache) {
				if (cache == null) {
					cache = new NodeCache(client, path, false);
					cache.start();
					listener.setCache(cache);
					listener.setSerializer(serializer);
					cache.getListenable().addListener(listener);
				}
			}
		}

		// TODO get from the cache?
		byte[] data = client.getData().watched().forPath(path);
		return serializer.deserialize(data);
	}

	public void delete(String path) throws Exception {
		try {
			client.delete().forPath(path);
		} catch (Exception ex) {
			if (!(ex instanceof NoNodeException)) {
				throw ex;
			}
		}
	}

	public List<String> getChildren(String path) throws Exception {
		/**
		 * Get children and set a watcher on the node. The watcher notification will
		 * come through the CuratorListener (see setDataAsync() above).
		 */

		return client.getChildren().forPath(path);
	}

	public List<String> getChildrenAndWatch(String path, ChildrenChangedListener listener) throws Exception {
		/**
		 * Get children and set a watcher on the node. The watcher notification will
		 * come through the CuratorListener (see setDataAsync() above).
		 */

		PathChildrenCache cache = path2ChildrenCache.get(path);
		if (cache == null) {
			synchronized (path2ChildrenCache) {
				if (cache == null) {
					cache = new PathChildrenCache(client, path, true, false, executor);
					cache.start();
					listener.setSerializer(serializer);
					cache.getListenable().addListener(listener);
				}
			}
		}

		return client.getChildren().watched().forPath(path);
	}

	public void removeListener(String path, ChildrenChangedListener listener) {
		PathChildrenCache cache = path2ChildrenCache.get(path);
		if (cache == null) {
			return;
		}

		cache.getListenable().removeListener(listener);
	}

	public boolean exist(String path) throws Exception {
		Stat s = client.checkExists().forPath(path);
		return s != null;
	}

	/**
	 * 选主
	 * @param path
	 * @param listener
	 * @throws Exception
	 */
	public void selectLeader(String path, String id, LeaderLatchListener listener) throws Exception {
		final LeaderLatch leaderLatch = new LeaderLatch(client, path, id);
		leaderLatch.addListener(listener);
		leaderLatch.start();
	}

	public CuratorFramework getClient(){
		return client;
	}


}